package drds.global_transaction.api.rpc;

import java.nio.ByteBuffer;
import java.util.List;

import drds.common.$;
import drds.common.Constants;
import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.message_codec.IMessageCodec;
import drds.global_transaction.api.message_codec.message_codec.MessageCodecs;
import drds.global_transaction.api.message_codec.rpc_message.RpcMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class MessageCodecHandler extends ByteToMessageCodec<RpcMessage>
{
	private static final int flag_request = 0x40;
	private static final int flag_async = 0x20;
	private static final int flag_heartbeat = 0x10;

	//
	private static final int not_found_index = -1;
	private static int head_length = 14;

	private static int getMagicIndex(ByteBuf byteBuf)
	{
		boolean found = false;
		int readIndex = byteBuf.readerIndex();
		int begin = 0;
		while (readIndex < byteBuf.writerIndex())
		{
			if (byteBuf.getByte(readIndex) == Magic.magic_half && byteBuf.getByte(readIndex + 1) == Magic.magic_half)
			{
				begin = readIndex;// 第一个magic_half为magic index
				found = true;
				break;
			}
			++readIndex;
		}
		return found ? begin : not_found_index;
	}

	@Override
	protected void encode(ChannelHandlerContext channelHandlerContext, RpcMessage rpcMessage, ByteBuf byteBuf) throws Exception
	{
		IMessageCodec messageCodec = null;
		ByteBuffer byteBuffer = ByteBuffer.allocate(128);
		if (rpcMessage.getMessageCodec() instanceof IMessageCodec)
		{
			messageCodec = (IMessageCodec) rpcMessage.getMessageCodec();
		}
		byteBuffer.putShort(Magic.magic);
		int flag = (rpcMessage.isAsync() ? flag_async : 0) | (rpcMessage.isHeartbeat() ? flag_heartbeat : 0) | (rpcMessage.isRequest() ? flag_request : 0);

		byteBuffer.putShort((short) flag);
		// byteBuffer to netty byteBuf
		if (rpcMessage.getMessageCodec() instanceof HeartbeatMessage)
		{
			byteBuffer.putShort((short) 0);
			byteBuffer.putLong(rpcMessage.getId());
			byteBuffer.flip();
			//
			byte[] bytes = new byte[byteBuffer.limit()];
			byteBuffer.get(bytes);// 写入数组中
			// netty write
			byteBuf.writeBytes(bytes);
			return;
		}

		try
		{
			if (null != messageCodec)
			{
				byteBuffer.putShort(messageCodec.getTypeCode());
				byteBuffer.putLong(rpcMessage.getId());
				byteBuffer.flip();
				//
				byte[] bytes = new byte[byteBuffer.limit()];
				byteBuffer.get(bytes);
				byteBuf.writeBytes(bytes);// 写入数组中
				// netty write
				byteBuf.writeBytes(messageCodec.encode());
			} else
			{
				throw new IllegalStateException();
			}
		} catch (Exception e)
		{
			log.error(rpcMessage.getMessageCodec() + " encode error", "", e);
			throw e;
		}
	}

	/**
	 * @param byteBuf    输入
	 * @param objectList 输出
	 */
	@Override
	protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> objectList) throws Exception
	{
		int begin = byteBuf.readerIndex();
		int magicIndex = getMagicIndex(byteBuf);
		if (magicIndex == not_found_index)
		{
			log.error("codec decode not found magic offset");
			byteBuf.skipBytes(byteBuf.readableBytes());
			return;
		}
		if (magicIndex != 0 && log.isInfoEnabled())
		{
			log.warn("please notice magicIndex is not zero offset!!!");
		}
		byteBuf.skipBytes(magicIndex - byteBuf.readerIndex());
		if (byteBuf.readableBytes() < head_length)
		{
			log.error("decode less than header length");
			return;
		}
		byte[] bytes = new byte[head_length];
		byteBuf.readBytes(bytes);
		// netty byteBuf to byteBuffer
		ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
		short magic = byteBuffer.getShort();
		if (magic != Magic.magic)
		{
			log.error("decode error,will close channel:" + channelHandlerContext.channel());
			channelHandlerContext.channel().close();
			return;
		}
		//
		int flag = byteBuffer.getShort();
		boolean heartbeat = (flag_heartbeat & flag) > 0;
		boolean isRequest = (flag_request & flag) > 0;
		//
		short typeCode = byteBuffer.getShort();
		//
		long messageId = byteBuffer.getLong();
		if (heartbeat)
		{
			RpcMessage rpcMessage = new RpcMessage();
			rpcMessage.setId(messageId);
			rpcMessage.setAsync(true);
			rpcMessage.setHeartbeat(heartbeat);
			rpcMessage.setRequest(isRequest);
			if (isRequest)
			{
				rpcMessage.setMessageCodec(HeartbeatMessage.ping);
			} else
			{
				rpcMessage.setMessageCodec(HeartbeatMessage.pong);
			}
			objectList.add(rpcMessage);
			return;
		}
		RpcMessage rpcMessage = new RpcMessage();
		rpcMessage.setId(messageId);
		rpcMessage.setAsync((flag_async & flag) > 0);
		rpcMessage.setHeartbeat(false);
		rpcMessage.setRequest(isRequest);
		try
		{
			IMessageCodec messageCodec = MessageCodecs.getMessageCodec(typeCode);
			if (!messageCodec.decode(byteBuf))
			{
				log.error(messageCodec + " decode error.");
				byteBuf.readerIndex(begin);
				return;
			}
			rpcMessage.setMessageCodec(messageCodec);
		} catch (Exception e)
		{
			if (Constants.developMode)
			{
				e.printStackTrace();
			}
			log.error($.printStackTraceToString(e));
			//
			log.error("decode error", "", e);
			throw e;
		}
		objectList.add(rpcMessage);

	}
}
